from pyspark import SparkConf, SparkContext
import os
import json

# 设置python解释器环境变量
os.environ['PYSPARK_PYTHON'] = 'D:/PYTHON/python3.10/python.exe'

conf = SparkConf().setMaster('local[*]').setAppName('my_test_spark')
sc = SparkContext(conf=conf)

# 数据输入
rdd = sc.textFile('E:/PythonBasicKnowledge/12 pyspark使用/orders.txt')

# 【1】城市销售额排名
# 1.1 把文件中的JSON数据字符串转换为Python中的dict类型
json_str_rdd = rdd.flatMap(lambda x: x.split('|'))  # 转换为json字符串
dict_rdd = json_str_rdd.map(lambda x: json.loads(x))  # 转换为字典
# 1.2 组成👉(城市名, 销售额)格式
city_money_rdd = dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
# 1.3 分组累加销售额
r1 = city_money_rdd.reduceByKey(lambda x, y: x + y).sortBy(lambda x: x[1], ascending=False, numPartitions=1)
print(F'【1】城市销售额排名：\n{r1.collect()}')

# 【2】全部城市有哪些商品类别在售卖
# 取出全部的商品类别并去重
r2 = dict_rdd.map(lambda x: x['category']).distinct()
print(F'【2】全部城市有哪些商品类别在售卖：\n{r2.collect()}')

# 【3】北京有哪些商品类别在售卖
beijing_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
# print(F'北京的数据：{beijing_rdd.collect()}')
r3 = beijing_rdd.map(lambda x: x['category']).distinct()
print(F'【3】北京有哪些商品类别在售卖：\n{r3.collect()}')

sc.stop()
